package com.cloud.easy.im.utils;

import com.cloud.easy.im.argeement.Agreement;
import com.cloud.easy.im.argeement.AgreementFactory;
import com.cloud.easy.im.argeement.AgreementType;
import com.cloud.easy.im.argeement.ErrorCode;
import com.cloud.easy.im.handler.qos.QoS4SendDaemonS2C;
import com.cloud.easy.im.netty.IMObserver;
import com.cloud.easy.im.processor.OnlineProcessor;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 基于本地消息发送
 *
 * @author wangjian
 * @version 0.1.0
 * @create 2020年09月15日4:28 下午
 */
public class LocalSendHelper {

    private static final Logger logger = LoggerFactory.getLogger(LocalSendHelper.class);

    /**
     * 发送消息的客户端尚未登陆
     *
     * @param session   接受人channel
     * @param agreement 消息体
     * @param observer  观察者回调
     */
    public static void replyDataForUnlogined(Channel session, Agreement agreement
            , IMObserver observer) throws Exception {
        logger.warn("[LOCAL_SEND_HELPER#replyDataForUnlogined]>> 客户端[" + agreement.getFrom() + "]尚未登陆，" + agreement.getDataContent() + "处理未继续.");

        Agreement perror = AgreementFactory.createPErrorResponse(
                // 尚未登陆则user_id就不存在了,用-1表示吧，目前此情形下该参数无意义
                ErrorCode.ForS.RESPONSE_FOR_UNLOGIN, agreement.toJsonString(), "-1",agreement.getProductCode());
        sendData(session, perror, observer);
    }

    /**
     * 发送消息到客户端的伪应答包
     *
     * @param session   接受人channel
     * @param agreement 消息体
     * @param observer  观察者回调
     * @throws Exception
     */
    public static void replyDelegateRecievedBack(Channel session, Agreement agreement
            , IMObserver observer) throws InterruptedException {
        if (agreement.isQoS() && agreement.getFp() != null) {
            Agreement receivedBackP = AgreementFactory.createRecivedBack(
                    agreement.getTo()
                    , agreement.getFrom()
                    , agreement.getFp()
                    ,agreement.getProductCode());

            sendData(session, receivedBackP, observer);
        } else {
            logger.warn("[LOCAL_SEND_HELPER#replyDelegateRecievedBack]收到userId:" + agreement.getFrom()+"|productCode:"+agreement.getProductCode()
                    + "发过来需要QoS的包，但它的指纹码却为null！无法发伪应答包！");

            if (observer != null) {
                observer.update(false, null);
            }
        }
    }

    /**
     * 发送消息
     * note：未确定收件人，需要从在线列表中获取，获取不到考虑离线保存
     *
     * @param agreement 请求体
     * @param observer  回调监听
     */
    public static void sendData(final Agreement agreement, final IMObserver observer) throws InterruptedException {
        if (agreement != null) {
            if (!Agreement.serverMark.equals(agreement.getTo())) {
                sendData(OnlineProcessor.getInstance().getOnlineSession(agreement.getTo(), agreement.getProductCode()), agreement, observer);
                return;
            } else {
                logger.warn("[LOCAL_SEND_HELPER]【注意】此Agreement对象中的接收方是服务器(user_id==0)（而此方法本来就是由Server调用，自已发自已不可能！），数据发送没有继续！" + agreement.toJsonString());
            }
        }
        if (observer != null) {
            observer.update(false, null);
        }

    }

    /**
     * 异步发送消息发送消息
     * note：已确定收件人，需判断是否在线
     *
     * @param session   接受人channel
     * @param agreement 消息体
     * @param observer  观察者回调
     *                  false：用户不在线/消息发送失败（推荐对消息做离线处理）
     *                  true：消息发送成功
     */
    public static void sendData(final Channel session
            , final Agreement agreement, final IMObserver observer) throws InterruptedException {
        if (session == null) {
            logger.info("[LOCAL_SEND_HELPER] ChannelSession==null >> id=" + agreement.getFrom() + "的用户尝试发给客户端" + agreement.getTo()
                    + "的消息：msg=" + agreement.getDataContent() + "因接收方的id已不在线(推荐做离线处理).");
        } else {
            //对方在线
            if (session.isActive()) {
                if (agreement != null) {
                    //异步监听
                    ChannelFuture cf = writeAndFlushByWS(session, agreement).sync();

                    cf.addListener(new ChannelFutureListener() {
                        @Override
                        public void operationComplete(ChannelFuture future) {
                            // 群聊暂不支持qos机制
                            if (future.isSuccess()) {
                                if (AgreementType.C.FROM_CLIENT_TYPE_OF_COMMON$PUSH == agreement.getType()) {
                                    //qos机制，将发送者发送数据存入队列
                                    if (agreement.isQoS() && !QoS4SendDaemonS2C.getInstance().hasRecieved(agreement.getFp())) {
                                        QoS4SendDaemonS2C.getInstance().put(agreement);
                                    }
                                }
                            } else {
                                logger.warn("[LOCAL_SEND_HELPER]给客户端的数据->" + agreement.toJsonString() + ",发送失败! (推荐做离线处理).");
                            }
                            if (observer != null) {
                                observer.update(future.isSuccess(), null);
                            }
                        }
                    });
                    return;
                }
            }
            logger.warn("[LOCAL_SEND_HELPER]客户端id=" + agreement.getFrom() + "要发给客户端" + agreement.getTo()
                    + "的实时消息：str=" + agreement.getDataContent() + "没有继续(此消息应考虑作离线处理哦).");
        }

        if (observer != null) {
            observer.update(false, null);
        }
    }

    /**
     * UDP 的底层调用方法
     *
     * @param session
     * @param agreement
     * @return
     */
    private static ChannelFuture writeAndFlushByUDP(final Channel session, Agreement agreement) {
        final byte[] res = agreement.toBytes();

        ByteBuf to = Unpooled.copiedBuffer(res);
        return session.writeAndFlush(to);
    }

    /**
     * webSocket 的底层调用方法
     *
     * @param session
     * @param agreement
     * @return
     */
    private static ChannelFuture writeAndFlushByWS(final Channel session, Agreement agreement) {
        TextWebSocketFrame resp = new TextWebSocketFrame(agreement.toJsonString());
        return session.writeAndFlush(resp);
    }

}
